package com.casstime.ec.cloud.discount.jms;

import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.stereotype.Component;

import com.casstime.commons.utils.JsonUtil;

/**
 * ClassName: JmsMessageProducer 
 * Function: (消息消费)
 * Reason: ADD REASON(可选).
 *  
 * @author wu.chen 
 * 2018年3月26日 上午9:44:13
 * @version   
 * @since JDK 1.8
 */
@Component
public class JmsMessageProducer {
	
	
	private static ConcurrentHashMap<String, ActiveMQTopic> topicPool = new ConcurrentHashMap<String, ActiveMQTopic>();

	private static ConcurrentHashMap<String, ActiveMQQueue> queuePool = new ConcurrentHashMap<String, ActiveMQQueue>();

	private  Logger logger = LoggerFactory.getLogger(JmsMessageProducer.class);
	
	@Autowired
	private JmsTemplate jmsTemplate;
	
	/**
	 * 发送Topic消息并接收响应，响应未返回时会阻塞.
	 * 
	 * @param topicName
	 *            Topic队列名
	 * @param message
	 *            消息体
	 * @return
	 * @throws JMSException
	 */
	public String sendAndReceiveTopicMessage(final String topicName, final String message) throws JMSException {

		Message replyMsg = jmsTemplate.sendAndReceive(topicFactory(topicName), new MessageCreator() {
			@Override
			public Message createMessage(Session session) throws JMSException {
				return session.createTextMessage(message);
			}
		});
		TextMessage textMessage = (TextMessage) replyMsg;
		return textMessage.getText();
	}
	
	/**
	 * 发送Topic消息并接收响应，响应未返回时会阻塞.
	 * 
	 * @param topicName
	 *            Topic队列名
	 * @param message
	 *            消息体
     * @param prop
	 *            消息体附加属性
	 * @return
	 * @throws JMSException
	 */
	public String sendAndReceiveTopicMessage(final String topicName, final String message, Map<String, Object> prop) throws JMSException {
		
		Message replyMsg = jmsTemplate.sendAndReceive(topicFactory(topicName), new MessageCreator() {
			@Override
			public Message createMessage(Session session) throws JMSException {
				Message msg = session.createTextMessage(message);
				Set<Entry<String, Object>> set = prop.entrySet();         
				Iterator<Entry<String, Object>> i = set.iterator();         
				while (i.hasNext()) {      
				     Map.Entry<String, Object> entry = i.next();    
				     if (entry.getValue() instanceof String) {
				    	 msg.setStringProperty(entry.getKey(), (String) entry.getValue());
				     } else if (entry.getValue() instanceof Integer) {
				    	 msg.setIntProperty(entry.getKey(), (Integer) entry.getValue());
				     } else if (entry.getValue() instanceof Long) {
				    	 msg.setLongProperty(entry.getKey(), (Long) entry.getValue());
				     } else if (entry.getValue() instanceof Double) {
				    	 msg.setDoubleProperty(entry.getKey(), (Double) entry.getValue());
				     } else if (entry.getValue() instanceof Float) {
				    	 msg.setFloatProperty(entry.getKey(), (Float) entry.getValue());
				     } else if (entry.getValue() instanceof Short) {
				    	 msg.setShortProperty(entry.getKey(), (Short) entry.getValue());
				     } else {
				    	 logger.warn("消息推送[消息体附加属性]为非基础类型,-->{}", prop);
				     }
				}   
				return msg;
			}
		});
		TextMessage textMessage = (TextMessage) replyMsg;
		return textMessage.getText();
	}

	/**
	 * 发送Topic消息.
	 * 
	 * @param topicName
	 *            Topic队列名
	 * @param message
	 *            消息体
	 * @return
	 * @throws JMSException
	 */
	public void sendTopicMessage(final String topicName, final String message) throws JMSException {

		jmsTemplate.send(topicFactory(topicName), new MessageCreator() {
			@Override
			public Message createMessage(Session session) throws JMSException {
				return session.createTextMessage(message);
			}
		});
	}
	
	/**
	 * 发送Topic消息.
	 * 
	 * @param topicName
	 *            Topic队列名
	 * @param message
	 *            消息体
	 * @param prop
	 *            消息体附加属性
	 * @return
	 * @throws JMSException
	 */
	public void sendTopicMessage(final String topicName, final String message, Map<String, Object> prop) throws JMSException {
		jmsTemplate.send(topicFactory(topicName), new MessageCreator() {
			@Override
			public Message createMessage(Session session) throws JMSException {
				Message msg = session.createTextMessage(message);
				Set<Entry<String, Object>> set = prop.entrySet();         
				Iterator<Entry<String, Object>> i = set.iterator();         
				while (i.hasNext()) {      
				     Map.Entry<String, Object> entry = i.next();    
				     if (entry.getValue() instanceof String) {
				    	 msg.setStringProperty(entry.getKey(), (String) entry.getValue());
				     } else if (entry.getValue() instanceof Integer) {
				    	 msg.setIntProperty(entry.getKey(), (Integer) entry.getValue());
				     } else if (entry.getValue() instanceof Long) {
				    	 msg.setLongProperty(entry.getKey(), (Long) entry.getValue());
				     } else if (entry.getValue() instanceof Double) {
				    	 msg.setDoubleProperty(entry.getKey(), (Double) entry.getValue());
				     } else if (entry.getValue() instanceof Float) {
				    	 msg.setFloatProperty(entry.getKey(), (Float) entry.getValue());
				     } else if (entry.getValue() instanceof Short) {
				    	 msg.setShortProperty(entry.getKey(), (Short) entry.getValue());
				     } else {
				    	 logger.warn("消息推送[消息体附加属性]为非基础类型,-->{}", prop);
				     }
				}   
				logger.debug("消息推送:--->{}", JsonUtil.serializer(msg));
				return msg;
			}
		});
	}

	/**
	 * 发送Queue消息并接收响应，响应未返回时会阻塞.
	 * 
	 * @param queueName
	 *            queue队列名
	 * @param message
	 *            消息体
	 * @return
	 * @throws JMSException
	 */
	public String sendAndReceiveQueueMessage(final String queueName, final String message) throws JMSException {
		Message replyMsg = jmsTemplate.sendAndReceive(queueFactory(queueName), new MessageCreator() {
			@Override
			public Message createMessage(Session session) throws JMSException {
				return session.createTextMessage(message);
			}
		});
		TextMessage textMessage = (TextMessage) replyMsg;
		return textMessage.getText();
	}

	/**
	 * 发送Queue消息.
	 * 
	 * @param queueName
	 *            queue队列名
	 * @param message
	 *            消息体
	 * @return
	 * @throws JMSException
	 */
	public void sendQueueMessage(final String queueName, final String message) throws JMSException {
		jmsTemplate.send(queueFactory(queueName), new MessageCreator() {
			@Override
			public Message createMessage(Session session) throws JMSException {
				return session.createTextMessage(message);
			}
		});
	}

	private ActiveMQTopic topicFactory(String topicName) {
		if (!topicPool.containsKey(topicName)) {
			topicPool.put(topicName, new ActiveMQTopic(topicName));
		}
		return topicPool.get(topicName);
	}

	private ActiveMQQueue queueFactory(String queueName) {
		if (!queuePool.containsKey(queueName)) {
			queuePool.put(queueName, new ActiveMQQueue(queueName));
		}
		return queuePool.get(queueName);
	}

}